package repeater

import (
    "github.com/go-redis/redis"
    "../common"
    "fmt"
    "strings"
    "strconv"
    "encoding/json"
)

type RedisWriter struct {
    workingOffset   int64
    committedOffset int64
    rc              *RepeaterController
    redis           *redis.Client
    redisName       string
    addrs           []string
    passwd          string
    db              int
    cacheSize       int
}

func NewRedisWriter(d map[string]string) (*RedisWriter, error) {
    redisName := d["redis_name"]
    if redisName == "" {
        return nil, fmt.Errorf("Wrong Config redis_name")
    }
    passwd := d["passwd"]
    db := d["db"]
    var dbid int
    dbid, _ = strconv.Atoi(db)
    s := d["redis_addrs"]
    if s == "" {
        return nil, fmt.Errorf("Wrong Config redis_addrs")
    }
    addrs := strings.Split(s, ",")
    return &RedisWriter{
        redisName: redisName,
        addrs:     addrs,
        passwd:    passwd,
        db:        dbid,
        cacheSize: 30,
    }, nil
}

func (ww *RedisWriter) Init() error {
    ww.redis = redis.NewFailoverClient(&redis.FailoverOptions{
        MasterName:    ww.redisName,
        SentinelAddrs: ww.addrs,
        Password:      ww.passwd,
        MaxRetries:    10,
        DB:            ww.db,
    })
    pong := ww.redis.Ping()
    return pong.Err()
}

func (ww *RedisWriter) Start() {
}
func (ww *RedisWriter) Stop() (err error) {
    return ww.redis.Close()
}
func (ww *RedisWriter) Name() string {
    return ww.redisName
}
func (ww *RedisWriter) Handle(dps *DataPointPack) error {
    cache := make(map[string]map[string][]float64)
    for _, dp := range dps.Datapoints {
        k1, k2, v := _keyValue(dp)
        if k1 == "" || k2 == "" {
            continue
        }
        if _, ok := cache[k1]; !ok {
            cache[k1] = make(map[string][]float64)
        }
        cache[k1][k2] = append(cache[k1][k2], v)
    }
    for k, v := range cache {
        s, err := json.Marshal(v)
        if err != nil {
            continue
        }
        r := ww.redis.RPush(k, s);
        if err := r.Err(); err != nil {
            log.Error("redis error is %v", err)
            return err
        }
    }
    ww.workingOffset = dps.Offset()
    return nil
}

func (ww *RedisWriter) SetController(rc *RepeaterController) error {
    ww.rc = rc
    return nil
}

func (ww *RedisWriter) Commit() (int64, error) {
    ww.committedOffset = ww.workingOffset
    return ww.workingOffset, nil
}

func _keyValue(dp *common.DataPoint) (k1, k2 string, v float64) {
    ret := common.Extract(dp.Name)
    day := common.Day(dp.Time)
    ip := ret["ip"]
    head := ret["head"]
    sub := ret["sub"]
    p := ret["path"]
    v = dp.Val
    if ip == "" || head == "" || sub == "" || day == "" {
        return
    }
    if p == "" {
        k1 = strings.Join([]string{"golia", ip, day, head}, ".")
        k2 = sub
        return
    }
    if strings.Contains(head, "disk") {
        if p1, err := common.Decode_base64(p); err == nil {
            p = p1
        }
    }
    k1 = strings.Join([]string{"golia", ip, day, head}, ".")
    k2 = fmt.Sprintf("%s.%s", sub, p)
    return
}
